package com.gin.monitor.agg.warn;

import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.*;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.time.Duration;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;


/**
 * 违法车辆分析
 *
1620637885000,0002,75312,京L22188,190.1,27,31
1620637886000,0002,75312,京L22188,190.1,27,31
1620637889000,0002,75312,京L22188,190.1,27,31
1620637999000,0002,75312,京L22189,190.1,27,31
 *
 * @author gin
 * @date 2021/4/30
 */
public class CarTrackInfoAnalysis {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //第一种，当前的违法车辆（在5分钟内）如果已经出警了。（最后输出道主流中做删除处理）。
        // nc -lk 8888
        SingleOutputStreamOperator<TrackInfo> stream = env.socketTextStream("node01", 8888)
                .map(new RichMapFunction<String, TrafficInfo>() {
                    @Override
                    public TrafficInfo map(String line) {
                        //数据转换为VO对象
                        String[] s = line.split(",");
                        return new TrafficInfo(Long.parseLong(s[0]), s[1], s[2], s[3], Double.parseDouble(s[4]), s[5], s[6]);
                    }
                })
                //自定义的过滤函数类，把违法车辆信息留下，其他都去掉
                .filter(new RichFilterFunction<TrafficInfo>() {
                    Connection conn = null;
                    PreparedStatement pst = null;
                    HashMap<String, ViolationInfo> map = new HashMap<>();
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //open 在计算程序初始化时调用
                        conn = DriverManager.getConnection("jdbc:mysql://node01/test", "root", "123456");
                        pst = conn.prepareStatement("select car ,violation, create_time from t_violation_list");
                        ResultSet set = pst.executeQuery();
                        while (set.next()) {
                            //缓存卡口信息
                            ViolationInfo info = new ViolationInfo(set.getString(1), set.getString(2),
                                   set.getLong(3));
                            map.put(info.car, info);
                        }
                        set.close();
                        pst.close();
                        conn.close();
                    }

                    @Override
                    public boolean filter(TrafficInfo trafficInfo) throws Exception {
                        ViolationInfo o = map.get(trafficInfo.car);
                        return null != o;
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficInfo>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                        .withTimestampAssigner((event, timestamp) -> event.actionTime))
                .map((MapFunction<TrafficInfo, TrackInfo>) info -> new TrackInfo(info.car,info.actionTime,
                        info.monitorId,info.roadId,info.areaId,info.speed));

        //批量写入数据到 sink 中，启动CountWindow 来完成批量插入，批量的条数由窗口大小决定
        //开窗
        stream.countWindowAll(10)
                .apply(new AllWindowFunction<TrackInfo, List<TrackInfo>, GlobalWindow>() {
                    @Override
                    public void apply(GlobalWindow window, Iterable<TrackInfo> input, Collector<List<TrackInfo>> out) throws Exception {
                        LinkedList<TrackInfo> infoList = new LinkedList<>();
                        for (TrackInfo trackInfo : input) {
                            infoList.add(trackInfo);
                        }
                        out.collect(infoList);
                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
